package com.bossien.flink.config;

import com.bossien.flink.common.JSimpleStringSchema;
import com.bossien.flink.interfaces.IExecutionHandlerConfig;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author 胡春林
 * 流式数据执行器
 */
@Slf4j
@Data
public abstract class RMQExcutionStreamHandlerConfig<T> implements IExecutionHandlerConfig<T> {

    private StreamExecutionEnvironment streamExecutionEnvironment;

    @Autowired
    private StreamEnvironmention streamEnvironmention;
    @Autowired
    private RabbitParamsConfig rabbitParamsConfig;
    @Autowired
    private JSimpleStringSchema jSimpleStringSchema;

    @Override
    public DataStream<String> addSource() throws IOException {

        //获取配置后的流环境
        StreamExecutionEnvironment executionEnvironment =  streamEnvironmention.getStreamExecutionEnvironment();
        //具体业务参数可在各自handler中配置（仅一次或至少一次，检查点是必须的）
        executionEnvironment.enableCheckpointing(5000);
        //单独设置针对该任务 也可以全局设置(并发设置为1 仅一次所必须的)
        executionEnvironment.setParallelism(1);
        executionEnvironment.setStateBackend(new FsStateBackend("file:///E:\\ceshi\\checkpoints"));
        log.info("获取配置后的环境成功！");
        DataStream<String> dataStream = executionEnvironment.addSource(new RMQSource<String>(rabbitParamsConfig.rmqConnectionConfig(),"flinkdata",false,jSimpleStringSchema));
        setStreamExecutionEnvironment(executionEnvironment);
        return dataStream;
    }

    @Override
    public abstract T executeFilter(T t);

    @Override
    public abstract T addSkin(T tClass);


    /**
     * 被代理类
     */
    @Component
    public  class RMQExcutionStreamHandlerConfigImpl extends RMQExcutionStreamHandlerConfig {

        @Override
        public Object executeFilter(Object o) { return null; }

        @Override
        public Object addSkin(Object tClass) { return null; }

    }
}
